package profile.dsplog2

import org.apache.commons.lang3.StringUtils
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}

import scala.collection.mutable

/**
  * Created by hunter.coder 涛哥  
  * 2019/4/17 15:42
  * 交流qq:657270652
  * Version: 1.0
  * 更多学习资料：https://blog.csdn.net/coderblack/
  * Description:  ID-MAPPING  id打通  识别出相同设备
  * 计算逻辑：
  * 1.从dsp日志数据中抽取每条数据中的各类id标识
  * 2.将每一个id标识转换成图计算中的"点"（vertex），去重后得到一个点集合
  * 3.将每条数据中的各标识（点），构建两两之间的“边”（edge），然后求相同边出现的次数，过滤掉次数低于指定阈值的，得到“边集合”
  * 4.将点集合和边集合构造成“图”（Graph）
  * 5.对图求“连通子图”  (点1->最小点1，点2->最小点1，点3->最小点1，点4->最小点4,点5->最小点4)
  * 6.根据连通子图结果得到id-mapping字典数据（进行存储）
  * 7.利用字典数据对dsp原始数据追加统一图id（逻辑：根据原始数据中的id去idmapping字典中逆向查询gid，而且决策依据：多胜少）
  **/
object IdMapping {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().master("local[*]").appName("IdMapping").getOrCreate()
    import spark.implicits._


    val inpath = "G:\\testdata\\usertags\\dspinput\\1.dat"

    // 使用真实数据
    /*val dspLogDf = spark.read.parquet("g:/sharkdata/dsppreout/")
    val ids_df = dspLogDf.select("imei","idfa","androidid","openudid","mac","uuid")*/

    // 加载dsp日志数据（使用测试数据（简化了一些字段））
    val dspLogDS = spark.read.textFile(inpath)
    //dspLogDS.cache()

    // 提取日志中的id标识
    val ids_df = dspLogDS.map(line => {
      val splits = line.split(",")
      val ids = splits(0).split(":", -1)
      (ids(0), ids(1), ids(2))
    }).toDF("imei", "idfa", "androidid")
    ids_df.cache()


    // 构造边集合
    val edgeRdd: RDD[Edge[Int]] = ids_df.rdd.flatMap(row => {
      val imei = row.getAs[String]("imei")
      val idfa = row.getAs[String]("idfa")
      val androidid = row.getAs[String]("androidid")

      val idsArr = Array(imei, idfa, androidid)

      // 过滤掉为空串的id码
      val idsArr2 = for (code <- idsArr if StringUtils.isNotBlank(code)) yield  code

      var tupleList: List[Edge[Int]] = Nil

      // 双重for循环，构造两两边
      for (i <- 0 to idsArr2.length - 2; j <- i + 1 to idsArr2.length - 1; if idsArr2.length > 1) {
        tupleList :+= Edge(idsArr2(i).hashCode().toLong, idsArr2(j).hashCode().toLong, 1)
      }

      tupleList

    })


    // 聚合边集合中，相同边出现的次数，并过滤掉次数低于3的边
    val aggEdgeStr: RDD[(String, Int)] = edgeRdd
      // 把每一个边对象变成(字符串,1)
      .map(edge=>(edge.srcId+"->"+edge.dstId,1))
      .reduceByKey(_+_)
      .filter(_._2>2)


    // 将过滤后的“字符串边”集合，转回Edge对象边
    val filteredEdges: RDD[Edge[String]] = aggEdgeStr.map(tp => {
      val split = tp._1.split("->")
      Edge(split(0).toLong, split(1).toLong, "")
    })


    // 构造一个点集合
    val idsVertices: Dataset[(Long, String)] = ids_df.flatMap(row=>{
      val imei = row.getAs[String]("imei")
      val idfa = row.getAs[String]("idfa")
      val androidid = row.getAs[String]("androidid")

      val idsArr = Array(imei, idfa, androidid)

      for (code <- idsArr if StringUtils.isNotBlank(code)) yield  code
    })
      // 对相同点去重
      .distinct()
      // 转换成"点"集合【Graph中的点就是一个tuple2(long值,任意属性)】
      .map(code=>(code.hashCode.toLong,code))


    //vetices.take(10).foreach(println)

    // 构造图
    val graph = Graph(idsVertices.rdd,filteredEdges)
    // 连通子图点
    val subVertices = graph.connectedComponents().vertices

    // 颠倒子图点的顺序（最小点id,点id）
    // （最小点01, List(id01)）  （最小点01, List(id02)）  （最小点01, List(id03)）
    val aggIds = subVertices.map(v => (v._2, v._1 :: Nil))
      // （最小点01, List(id01,id02,id03)）
      .reduceByKey(_ ++ _)

    aggIds.cache()


    // 保存idmapping的结果为文本文件
    aggIds.map(tp=>tp._1+"\001"+tp._2.mkString(","))
      .coalesce(1)
      .saveAsTextFile("G:\\testdata\\usertags\\idmapping-T")


    /**
      * 为DSP日志数据打上统一图id（gid）
      * 按道理，这个操作应该另外写一个程序
      */
    // 将连通子图点，收集到driver端
    val idsDict = subVertices.map(tp=>(tp._1.toString.toLong,tp._2.toString.toLong)).collectAsMap()


    // 将ids字典广播出去
    val bc = spark.sparkContext.broadcast(idsDict)


    // 对原始日志逐行打唯一标识
    dspLogDS.map(line=>{
      val splits = line.split(",")
      val ids = splits(0).split(":", -1)
      // 根据日志行中的各id去逆向查询统一图id
      val guid = getGuid(ids,bc.value)
      // 给原日志数据追加统一图id返回
      guid +"," + line
    }).coalesce(1)
      // 转成parquet存储（以便让这份测试数据和真实数据的处理形式保持一致）
      .map(line=>{
        val split = line.split(",", -1)
        val ids = split(1).split(":", -1)
        (split(0), ids(0), ids(1), ids(2), split(2))
      }).toDF("guid", "imei", "idfa", "androidid", "keywords")
      .write.parquet("G:\\testdata\\usertags\\dspidlog-T")


    spark.close()
  }


  // 根据各类标识码，逆向查找它所属的统一图id
  // 多个id标识可能逆向查询到多个图id，决策规则：少数服从多数
  def getGuid(ids: Array[String], idsMap: collection.Map[Long, Long]):Long = {

    // 用于记录guid被查到的次数
    val cntMap:mutable.HashMap[Long,Int] = new mutable.HashMap[Long,Int]()

    for (id <- ids) {
      val gid: Long = idsMap.getOrElse(id.hashCode.toLong,-1)
      if(gid.toLong != -1){
        cntMap.put(gid,cntMap.getOrElse(gid,0)+1)
      }
    }

    val list: List[(Long, Int)] = cntMap.toList.sortBy(-_._2)
    list(0)._1
  }

}
